package com.yeyks.rocketmq;

import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;
import com.yeyks.common.dto.mq.WebToBCnMessage;
import com.yeyks.common.utils.apputil.redis.RedisService;
import com.yeyks.common.utils.apputil.rediskey.BuilderKey;
import com.yeyks.common.utils.apputil.rediskey.RedisKey;
import com.yeyks.search.constants.EsIndexConstant;
import com.yeyks.search.service.ElasticSearchService;
import com.yeyks.search.service.hotel.EsImportHotelService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 处理搜索时的最新数据更新
 * @author dagny
 * @date 2019/12/28 0028 16:02
 */
@Slf4j
@Component
public class MessageCnBookingSearchListener implements BatchMessageListener {

    private final ElasticSearchService elasticSearchService;

    private final EsImportHotelService esImportHotelService;

    private final RedisService redisService;

    @Autowired
    public MessageCnBookingSearchListener(EsImportHotelService esImportHotelService,
                                          ElasticSearchService elasticSearchService,
                                          RedisService redisService) {
        this.esImportHotelService = esImportHotelService;
        this.elasticSearchService = elasticSearchService;
        this.redisService = redisService;
    }

   /* @Override
    public Action consume(Message message, ConsumeContext context) {
        try {
            //Long startTime = System.currentTimeMillis();
            byte[] body = message.getBody();
            //获取到接收的消息，由于接收到的是byte数组，所以需要转换成字符串
            String msg = new String(body);
            //处理业务逻辑
            this.disposeLogic(msg);
            //Long endTime = System.currentTimeMillis();
        } catch (Exception e) {
            log.error("MessageListener.consume error:" + e.getMessage(), e);
        }
        return Action.CommitMessage;
    }*/

    private void disposeLogic(String msg) {
        try {
            if(StringUtils.isNotEmpty( msg )){
                List<WebToBCnMessage> list = JSON.parseArray( msg, WebToBCnMessage.class );
                //循环更新信息（应该加一个机制，比如该酒店在1个小时内更新过，就直接跳过） key： sourceId  value： type + provinceId + : + cityId
                list.forEach( webToBCnMessage -> {
                    String key = BuilderKey.build( RedisKey.CN_HOTEL_UPDATED, webToBCnMessage.getSourceId());
                    if(StringUtils.isNotEmpty( redisService.get( key ) )){
                        return;
                    }
                    //清除酒店,房态
                    elasticSearchService.deleteCalendarByHotelId(webToBCnMessage.getSourceId(), EsIndexConstant.INDEX_HOTEL2);
                    //重新加载
                    esImportHotelService.importCnBookingHotelByHotelId(
                            Integer.parseInt(webToBCnMessage.getSourceId()),
                            EsIndexConstant.INDEX_HOTEL2,
                            "0001",
                            webToBCnMessage.getProvinceId(),
                            webToBCnMessage.getCityId());
                    //进行redis的存储
                    String value = webToBCnMessage.getSourceType() + ":" + webToBCnMessage.getProvinceId() + ":"
                            + webToBCnMessage.getCityId();
                    redisService.setAndExpire( key, value, 10, TimeUnit.MINUTES );
                } );
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("MessageCnBookingPush：处理监听消息异常：msg = {}", msg);
        }
    }

    @Override
    public Action consume(List<Message> messages, ConsumeContext context) {
        messages.forEach( message -> {
            try {
                //Long startTime = System.currentTimeMillis();
                byte[] body = message.getBody();
                //获取到接收的消息，由于接收到的是byte数组，所以需要转换成字符串
                String msg = new String(body);
                //处理业务逻辑
                this.disposeLogic(msg);
                //Long endTime = System.currentTimeMillis();
            } catch (Exception e) {
                log.error("MessageListener.consume error:" + e.getMessage(), e);
            }
        } );
        return Action.CommitMessage;
    }
}
